package api

import akka.stream.scaladsl.Source
import akka.{Done, NotUsed}
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.api.broker.kafka.{KafkaProperties, PartitionKeyStrategy}
import com.lightbend.lagom.scaladsl.api.transport.Method
import com.lightbend.lagom.scaladsl.api.{Descriptor, Service, ServiceCall}
import play.api.libs.json.{Format, Json}

/**
 * Created by yz on 11/3/2020
 */
object HelloWorldService {

  val TOPIC_NAME = "greetings"

}

trait HelloWorldService extends Service {

  def hello(id: String): ServiceCall[NotUsed, String]

  def useGreeting(id: String): ServiceCall[GreetingMessage, Done]

  def tick(interval: Int): ServiceCall[String, Source[String, NotUsed]]

  def greetingsTopic(): Topic[GreetingMessageChanged]

  override def descriptor: Descriptor = {
    import Service._
    named("hello-world").withCalls(
      pathCall("/api/hello/:id", hello _),
      pathCall("/api/hello/:id", useGreeting _),
      pathCall("/tick/:interval", tick _),
    ).withTopics(
      topic(HelloWorldService.TOPIC_NAME, greetingsTopic _).addProperty(
        KafkaProperties.partitionKeyStrategy,
        PartitionKeyStrategy[GreetingMessageChanged](_.name)
      )
    ).withAutoAcl(true)
  }

}

case class GreetingMessage(message: String)

object GreetingMessage {

  implicit val format: Format[GreetingMessage] = Json.format[GreetingMessage]

}

case class GreetingMessageChanged(name: String, message: String)

object GreetingMessageChanged {

  implicit val format: Format[GreetingMessageChanged] = Json.format[GreetingMessageChanged]

}


